package com.sqoop.commons.receive.impl;

import com.sqoop.commons.receive.MsgListener;
import com.sqoop.core.destnation.MqMsgDestination;
import com.sqoop.core.msg.Msg;
import com.sqoop.core.msg.MsgDestination;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQTopic;
import org.apache.commons.lang3.StringUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.jms.core.JmsTemplate;

import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by zw on 2017/11/10.
 */
public class MqMsgListener implements MsgListener {


    private static Logger logger = LogManager.getLogger(MqMsgListener.class.getName());

    //连接信息
    private Map<String,ReceiveMqClientInfo> clientInfoMap = new ConcurrentHashMap<>();

    // url  queue  两个级别
    private Map<String,Map<String, MqMsgReceiver>> msgReceiveMap = new ConcurrentHashMap<>();

    //已经存在的连接列表
    private Map<String,AtomicInteger> destinationMap = new ConcurrentHashMap<>();

    //运行的线程池
    private ExecutorService executorService = Executors.newCachedThreadPool();

    //消息汇总队列
    private LinkedBlockingQueue<Msg> linkedBlockingQueue = new LinkedBlockingQueue<>();

    private volatile boolean autoClean = false;
    //private Map<String,JmsTemplate>

    public class ReceiveMqClientInfo{
        private ConnectionFactory connectionFactory;
        private JmsTemplate jmsTemplate;

        public ReceiveMqClientInfo(ConnectionFactory connectionFactory, JmsTemplate jmsTemplate) {
            this.connectionFactory = connectionFactory;
            this.jmsTemplate = jmsTemplate;
        }

        public ConnectionFactory getConnectionFactory() {
            return connectionFactory;
        }

        public void setConnectionFactory(ConnectionFactory connectionFactory) {
            this.connectionFactory = connectionFactory;
        }

        public JmsTemplate getJmsTemplate() {
            return jmsTemplate;
        }

        public void setJmsTemplate(JmsTemplate jmsTemplate) {
            this.jmsTemplate = jmsTemplate;
        }
    }


    @Override
    public Msg getMsg() {
        try {
            return linkedBlockingQueue.take();
        } catch (InterruptedException e) {
            e.printStackTrace();
            logger.error("linkedBlockingQueue take error!!!" ,e);
        }
        return null;
    }



    @Override
    public void register(MsgDestination msgDestination) {
        if (msgDestination instanceof MqMsgDestination){
            MqMsgDestination mqMsgDestination = (MqMsgDestination)msgDestination;
            String url = mqMsgDestination.getMqurl();
            String username = StringUtils.isEmpty(mqMsgDestination.getUsername())? ActiveMQConnection.DEFAULT_USER:mqMsgDestination.getUsername();
            String passwd = StringUtils.isEmpty(mqMsgDestination.getPassword()) ? ActiveMQConnection.DEFAULT_PASSWORD :mqMsgDestination.getPassword();
            String queuetype = mqMsgDestination.getQueuetype();
            String queuename = mqMsgDestination.getQueuename();

            String key = getKey(url,queuename,queuetype);
            if (destinationMap.containsKey(key)){
                int hitsize = destinationMap.get(key).incrementAndGet();
                logger.info("key has registered!!! no need register twice!! hitsize is {}" ,hitsize);
                return;
            }
            Destination destination = null;
            if(queuetype.equalsIgnoreCase("queue")){
                destination = new ActiveMQQueue(queuename);
            }else{
                destination = new ActiveMQTopic(queuename);
            }
            destinationMap.put(key,new AtomicInteger());
            //如果连接池中存在连接
            if (this.clientInfoMap.containsKey(url)){
                MqMsgReceiver mqMsgReceiver = new MqMsgReceiver(clientInfoMap.get(url).getJmsTemplate(),destination,this);
                executorService.execute(mqMsgReceiver);

                //将消息接收放入容器
                if(!msgReceiveMap.containsKey(url)){
                    msgReceiveMap.put(url,new ConcurrentHashMap<>());
                }
                msgReceiveMap.get(url).put(queuename,mqMsgReceiver);


            }else {
                //新建连接 放入池中
                ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(username,passwd,url);
                connectionFactory.setTrustAllPackages(true);
                JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory);
                ReceiveMqClientInfo receiveMqClientInfo = new ReceiveMqClientInfo(connectionFactory,jmsTemplate);
                this.clientInfoMap.put(url,receiveMqClientInfo);
                //建立连接
                MqMsgReceiver mqMsgReceiver = new MqMsgReceiver(jmsTemplate,destination,this);
                executorService.execute(mqMsgReceiver);
                if(!msgReceiveMap.containsKey(url)){
                    msgReceiveMap.put(url,new ConcurrentHashMap<>());
                }
                msgReceiveMap.get(url).put(queuename,mqMsgReceiver);
            }
        }
    }

    @Override
    public void stop() {
        for (Map.Entry<String, Map<String, MqMsgReceiver>> entry:msgReceiveMap.entrySet()){
            for (Map.Entry<String, MqMsgReceiver> mqMsgReceiverentry:entry.getValue().entrySet()){
                mqMsgReceiverentry.getValue().stop();
            }
        }
        msgReceiveMap.clear();

    }


    public void stop(String url,String queueName){
        if (msgReceiveMap.containsKey(url) && msgReceiveMap.get(url).containsKey(queueName)){
            MqMsgReceiver msgReceiver = msgReceiveMap.get(url).get(queueName);
            if (msgReceiver != null){
                msgReceiver.stop();
            }
            //将信息从连接池中移除  但是这个移除工作
            msgReceiveMap.get(url).remove(queueName);
            if (autoClean) {
                clearUnusedConnection();
            }
        }
    }

    /**
     * 就是清除不用的连接
     * 判断clientInfoMap 以及 msgReceiveMap 的连接 ，如果有空闲使用的连接，关闭
     * 就是一个url在clientInfoMap存在 但是在msgReceiveMap不存在 ，那么就是无用的连接
     */
    private void clearUnusedConnection() {
    }


    @Override
    public void stop(Destination destination){
        if (destination instanceof MqMsgDestination){
            MqMsgDestination mqMsgDestination = (MqMsgDestination)destination;
            stop(mqMsgDestination.getMqurl(),mqMsgDestination.getQueuename());
        }
    }

    public void pause(String url,String queueName){
        if (msgReceiveMap.containsKey(url) && msgReceiveMap.get(url).containsKey(queueName)){
            MqMsgReceiver msgReceiver = msgReceiveMap.get(url).get(queueName);
            if (msgReceiver != null){
                msgReceiver.pause();
            }
            //暂停不需要移除
        }
    }

    @Override
    public void pause(Destination destination) {
        if (destination instanceof MqMsgDestination){
            MqMsgDestination mqMsgDestination = (MqMsgDestination)destination;
            pause(mqMsgDestination.getMqurl(),mqMsgDestination.getQueuename());
        }
    }

    @Override
    public void pause() {
        for (Map.Entry<String, Map<String, MqMsgReceiver>> entry:msgReceiveMap.entrySet()){
            for (Map.Entry<String, MqMsgReceiver> mqMsgReceiverentry:entry.getValue().entrySet()){
                mqMsgReceiverentry.getValue().pause();
            }
        }
    }


    private String getKey(Object ...objects){
        StringBuffer sb = new StringBuffer();
        for (Object obj:objects){
            sb.append(obj);
        }
        return sb.toString();
    }


    public void addMsg(Msg msg,String name){
        if (msg == null){
            logger.error("receive a error msg,sender is name!!" + name);
        }
        this.linkedBlockingQueue.add(msg);
    }

}